package com.atguigu.gmall.realtime.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.DruidUtil;
import com.atguigu.gmall.realtime.util.GmallConfig;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-09-09 22:52
 * <p>
 * 对mysql和kafka数据进行处理
 */
public class TableProcessFunc extends BroadcastProcessFunction<JSONObject, String, com.alibaba.fastjson.JSONObject> {


    private DruidDataSource dataSource;

    MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunc(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidUtil.createDataSource();
    }

    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, com.alibaba.fastjson.JSONObject>.ReadOnlyContext ctx, Collector<com.alibaba.fastjson.JSONObject> out) throws Exception {
        // 获取表名
        String tableName = value.getString("table");
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = broadcastState.get(tableName);
        //校验是否是维度表
        if (tableProcess != null) {
            //获取当前操作影响的业务记录
            JSONObject dataJsonObj = value.getJSONObject("data");

            //在向下游传递数据前，将不需要传递的属性过滤掉
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj, sinkColumns);

            //在向下游传递数据前，应该补充维度发送的目的地 sink_table
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table", sinkTable);

            //如果获取到了，说明当前处理的业务流数据是维度
            out.collect(dataJsonObj);
        } else {
            System.out.println("当前表数据不属于维度表，表名为" + tableName + "数据为" + value);
        }
    }

    @Override
    public void processBroadcastElement(String jsonStr, BroadcastProcessFunction<JSONObject, String, com.alibaba.fastjson.JSONObject>.Context ctx, Collector<com.alibaba.fastjson.JSONObject> out) throws Exception {
        // 对mysql 的维度数据进行处理
        JSONObject jsonObject = JSON.parseObject(jsonStr);

        TableProcess tableProcess = jsonObject.getObject("after", TableProcess.class);

        // 获取数据
        String sinkTable = tableProcess.getSinkTable();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkPk = tableProcess.getSinkPk();
        String sinkExtend = tableProcess.getSinkExtend();

        // 在 phonies 中创建表
        checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);

        //将维度数据放入广播变量中
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        broadcastState.put(tableProcess.getSourceTable(), tableProcess);
    }

    /**
     * 先在phoneis中创建表 前提创建schema
     *
     * @param tableName 表名
     * @param columnStr 列明
     * @param pk        主键
     * @param ext       扩展
     */
    public void checkTable(String tableName, String columnStr, String pk, String ext) {
        //对空值进行处理
        if (ext == null) {
            ext = "";
        }
        if (pk == null) {
            pk = "id";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + "(");

        String[] columnArr = columnStr.split(",");
        for (int i = 0; i < columnArr.length; i++) {
            String column = columnArr[i];
            if (column.equals(pk)) {
                createSql.append(column).append(" varchar primary key");
            } else {
                createSql.append(column).append(" varchar");
            }
            //除了最后一个字段后面不加逗号，其它的字段都需要在后面加一个逗号
            if (i < columnArr.length - 1) {
                createSql.append(",");
            }
        }
        createSql.append(") ").append(ext);
        System.out.println("在phoenix中执行的建表语句: " + createSql);
        // 在phoenix 生成表
        try {
            //从连接池中获取连接对象
            Connection conn = dataSource.getConnection();
            PhoenixUtil.executeSQL(createSql.toString(),conn);
        } catch (Exception e) {
            e.printStackTrace();
        }
        /*PreparedStatement ps = null;
        Connection conn = null;
        try {
            //从连接池中获取连接对象
            conn = dataSource.getConnection();
            //创建数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行sql语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }*/
    }

    /**
     * 过滤数据
     *
     * @param dataJsonObj 数据
     * @param sinkColumns 数据
     */
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        //方案一
        String[] fieldArr = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fieldArr);
        //获取json对象中所有的元素
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry -> !fieldList.contains(entry.getKey()));
        //方案二 遍历赋值
        /*JSONObject data = new JSONObject();
        String[] colums = sinkColumns.split(",");
        for (String colum : colums) {
            data.put(colum,dataJsonObj.getString(colum));
        }
        return data;*/
    }

}
